package com.chencong.online.function;

import com.chencong.online.bean.OrderEventBean;
import com.chencong.online.bean.ReceiptEventBean;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @program: user-behavior-analysis-online
 * @ClassName PayReceiptCoProcessFunc
 * @description:自定义两条流匹配函数
 * @author: chencong
 * @create: 2022-01-06 14:34
 **/
public class PayReceiptCoProcessFunc extends CoProcessFunction<OrderEventBean, ReceiptEventBean, Tuple2<OrderEventBean, ReceiptEventBean>> {
    //定义状态来保存当前来过的订单支付和到账时间
    ValueState<OrderEventBean> payState;
    ValueState<ReceiptEventBean> receiptState;

    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化状态
        payState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEventBean>("pay-state", OrderEventBean.class));
        receiptState = getRuntimeContext().getState(new ValueStateDescriptor<ReceiptEventBean>("receipt-state", ReceiptEventBean.class));
    }

    @Override
    public void processElement1(OrderEventBean pay, CoProcessFunction<OrderEventBean, ReceiptEventBean, Tuple2<OrderEventBean, ReceiptEventBean>>.Context ctx, Collector<Tuple2<OrderEventBean, ReceiptEventBean>> out) throws Exception {//
        //订单事件来了，判断是否有对应的到账事件
        ReceiptEventBean receipt = receiptState.value();
        if (receipt != null) {
            //如果receipt不为null，说明到账事件已经来过，输出主流，清空状态
            out.collect(new Tuple2<>(pay, receipt));
            payState.clear();
            receiptState.clear();
        } else {
            //如果receipt没来，注册一个定时器等待
            long ts = (pay.getTimestamp() + 5) * 1000L;//注册一个五秒钟的定时器
            ctx.timerService().registerEventTimeTimer(ts);
            //更新状态
            payState.update(pay);
        }
    }

    @Override
    public void processElement2(ReceiptEventBean receipt, CoProcessFunction<OrderEventBean, ReceiptEventBean, Tuple2<OrderEventBean, ReceiptEventBean>>.Context ctx, Collector<Tuple2<OrderEventBean, ReceiptEventBean>> out) throws Exception {
        //到账事件来了，判断是否有对应的订单数据
        OrderEventBean pay = payState.value();
        if (pay != null) {
            //如果pay不为null，说明订单事件已经来过，输出主流，清空状态
            out.collect(new Tuple2<>(pay, receipt));
            payState.clear();
            receiptState.clear();
        } else {
            //如果pay没来。注册一个定时器等待
            long ts = (pay.getTimestamp() + 3) * 1000L;//注册一个五秒钟的定时器
            ctx.timerService().registerEventTimeTimer(ts);
            //更新状态
            receiptState.update(receipt);
        }
    }

    @Override
    public void onTimer(long timestamp, CoProcessFunction<OrderEventBean, ReceiptEventBean, Tuple2<OrderEventBean, ReceiptEventBean>>.OnTimerContext ctx, Collector<Tuple2<OrderEventBean, ReceiptEventBean>> out) throws Exception {
        //定时器触发
        //判断哪一个不为空，另一个就没来
        if (payState.value() == null) {
            //此时说明pay没有来
            ctx.output(new OutputTag<Tuple2<OrderEventBean, ReceiptEventBean>>("pay没有来！！") {
            }, new Tuple2(payState.value(), receiptState.value()));
        }
        if (receiptState.value() == null) {
            //此时说明receipt没来
            ctx.output(new OutputTag<Tuple2<OrderEventBean, ReceiptEventBean>>("receipt没有来！！") {
            }, new Tuple2<>(payState.value(), receiptState.value()));
        }
        //清空状态
        payState.clear();
        receiptState.clear();
    }
}
